Make execution plan/blocklist aware of the memory ownership and who runs the plan #26650

jianoaix · 2022-07-17T22:41:47Z

Why are these changes needed?

Having the indicator about who's running the stage and who created a blocklist will enable the eager memory releasing.

This is an alternative with better abstraction to #26196.

Note: this doesn't work for Dataset.split() yet, will do in a followup PR.

Related issue number

#25249

Checks

I've run scripts/format.sh to lint the changes in this PR.
I've included any doc changes needed for https://docs.ray.io/en/master/.
I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
Testing Strategy
- Unit tests
- Release tests
- This PR is not tested :(

…tPipeline

ericl

This seems a lot more natural, though I'm wondering if we can improve the naming / concept a little bit more. The proposal seems OK to me though. What do others think?

python/ray/data/_internal/block_list.py

python/ray/data/_internal/plan.py

…lineplan

Signed-off-by: Ubuntu <[email protected]>

My experiments (the script ) showed that dataset.split_at_indices() with SPREAD tasks have more predictable performance Concretely: on 10 m5.4xlarge nodes with 5000 iops disk calling ds.split_at_indices(81) on 200GB dataset with 400 blocks: the split_at_indices without this PR takes 7-19 seconds, split_at_indices with SPREAD takes 7-12 seconds. Signed-off-by: Ubuntu <[email protected]>

…on (ray-project#26634) Signed-off-by: Ubuntu <[email protected]>

…ject#26458) Signed-off-by: Ubuntu <[email protected]>

Co-authored-by: Kourosh Hakhamaneshi <[email protected]> Signed-off-by: Ubuntu <[email protected]>

…ject#26094) Co-authored-by: Eric Liang <[email protected]> Co-authored-by: matthewdeng <[email protected]> Co-authored-by: Matthew Deng <[email protected]> Co-authored-by: Richard Liaw <[email protected]> Signed-off-by: Ubuntu <[email protected]>

…y-project#26643) Signed-off-by: Ubuntu <[email protected]>

…project#26637) Signed-off-by: Ubuntu <[email protected]>

…roject#26640) Signed-off-by: Ubuntu <[email protected]>

…ints (ray-project#26641) This PR replaces dataset.split(.., equal=True) implementation by dataset.split_at_indices() . My experiments (the script ) showed that dataset.split_at_indices() have more predictable performance than the dataset.split(…) Concretely: on 10 m5.4xlarge nodes with 5000 iops disk calling ds.split(81) on 200GB dataset with 400 blocks: the split takes 20-40 seconds, split_at_indices takes ~12 seconds. calling ds.split(163) on 200GB dataset with 400 blocks, the split takes 40-100 seconds, split_at_indices takes ~24 seconds. I don’t have much insight of dataset.split implementation, but with dataset.split_at_indices() we are just doing SPREAD to num_split_at_indices tasks, which yield much stable performance. Note: clean up the usage of experimental locality_hints in ray-project#26647 Signed-off-by: Ubuntu <[email protected]>

Why are these changes needed? Since locality_hints is an experimental feature, we stop promoting it in doc and don't enable it in AIR. See ray-project#26641 for more context Signed-off-by: Ubuntu <[email protected]>

Signed-off-by: Ubuntu <[email protected]>

…ray-project#26428) Signed-off-by: Ubuntu <[email protected]>

…project#26521) Signed-off-by: Ubuntu <[email protected]>

- Stop using dot command to run ci.sh script: it doesn't fail the build if the command fails for windows and is generally dangerous since it will make unexpected changes to the current shell. - Fix uncovered windows build issues. Signed-off-by: Ubuntu <[email protected]>

Signed-off-by: Ubuntu <[email protected]>

python/ray/data/_internal/block_list.py

python/ray/data/_internal/lazy_block_list.py

python/ray/data/_internal/plan.py

…lineplan

python/ray/data/_internal/block_list.py

python/ray/data/_internal/plan.py

jianoaix · 2022-07-21T00:04:59Z

@ericl @clarkzinzow For "run_by_consumer" that's currently attached to each stage, I'm thinking it makes more sense to be attached to ExecutionPlan instead, since that's the actual unit run by the consumer. Each individual stage can derive that from the ExecutionPlan it belongs to. I'll make it so if this sounds good to you.

…is arg in ExecutionPlan instead of Stage

c21 · 2022-07-21T05:02:55Z

python/ray/data/_internal/block_list.py

        assert len(blocks) == len(metadata), (blocks, metadata)
        self._blocks: List[ObjectRef[Block]] = blocks
        self._num_blocks = len(self._blocks)
        self._metadata: List[BlockMetadata] = metadata
+        # Whether the block list is owned by consuming APIs, and if so it can be
+        # eagerly deleted after read by the consumer.
+        self._owned_by_consumer = owned_by_consumer


shall we clearly define what is consumer in one place, e.g. dataset.py?

c21 · 2022-07-21T05:05:48Z

python/ray/data/_internal/block_list.py

+        blocks: List[ObjectRef[Block]],
+        metadata: List[BlockMetadata],
+        *,
+        owned_by_consumer: bool,


is it the idea that if the blocks are owned_by_consumer, then we can eagerly clean them up?

Yep, once it's consumed, the blocklist can be cleared.

c21 · 2022-07-21T05:08:09Z

python/ray/data/dataset.py

@@ -1058,16 +1058,20 @@ def equalize(splits: List[Dataset[T]], num_splits: int) -> List[Dataset[T]]:

        block_refs, metadata = zip(*blocks.get_blocks_with_metadata())
        metadata_mapping = {b: m for b, m in zip(block_refs, metadata)}
+        owned_by_consumer = blocks._owned_by_consumer


hmm I don't see iter_batches() set run_by_consumer, or any other consumer APIs (such as show(), take(), write_xxx). Am I missing anything?

We cannot do this for Dataset yet, for now it's unchanged.

c21 · 2022-07-21T05:12:27Z

I am wondering in which cases we do need eager execution for now? Can we remove support for eager execution post 2.0, and support lazy execution only? That would make code structure much simpler I guess. In that case, only the input and output blocks of execution plan needs be preserved, and all intermediate blocks can be cleaned up.

We could introduce another dataset persist()/cache() API to allow user run the execution plan and save the output blocks, in case they need.

jianoaix · 2022-07-21T05:25:54Z

I am wondering in which cases we do need eager execution for now? Can we remove support for eager execution post 2.0, and support lazy execution only? That would make code structure much simpler I guess. In that case, only the input and output blocks of execution plan needs be preserved, and all intermediate blocks can be cleaned up.

We could introduce another dataset persist()/cache() API to allow user run the execution plan and save the output blocks, in case they need.

Currently in call cases (except read) it's eager. I think lazy only (not just lazy by default) may make sense, just like DatasetPipeline. In that case, we can remove the "run_by_consumer", because it's always run by consumer.
We actually can also remove input (if it's recomputable, like read from external storage or computed via math operators like range()), and also output (e.g. in DatasetPipeline, we delete output once it's consumed).
We may just keep those pinned in memory, including in-memory data used to create initial input blocks, and those explicitly pinned. Today we have an API fully_executed(), which could serve similar purpose as cache API as you mentioned. In lazy only world, each consumption/action is recomputed from pinned memory or external storage.

…uns the plan (ray-project#26650) Having the indicator about who's running the stage and who created a blocklist will enable the eager memory releasing. This is an alternative with better abstraction to ray-project#26196. Note: this doesn't work for Dataset.split() yet, will do in a followup PR. Signed-off-by: Rohan138 <[email protected]>

…uns the plan (ray-project#26650) Having the indicator about who's running the stage and who created a blocklist will enable the eager memory releasing. This is an alternative with better abstraction to ray-project#26196. Note: this doesn't work for Dataset.split() yet, will do in a followup PR. Signed-off-by: Stefan van der Kleij <[email protected]>

Make execution stage/blocklist aware if it's run by Dataset or Datase…

6414e58

…tPipeline

jianoaix requested review from ericl, scv119, clarkzinzow and jjyao as code owners July 17, 2022 22:41

jianoaix added the @author-action-required The PR author is responsible for the next step. Remove tag to send back to the reviewer. label Jul 17, 2022

jianoaix mentioned this pull request Jul 17, 2022

Revert "Revert "Object GC for block splitting inside the dataset spli… #26583

Closed

6 tasks

ericl reviewed Jul 17, 2022

View reviewed changes

python/ray/data/_internal/block_list.py Outdated Show resolved Hide resolved

python/ray/data/_internal/plan.py Outdated Show resolved Hide resolved

Ubuntu and others added 20 commits July 18, 2022 16:59

Merge branch 'master' of https://github.com/ray-project/ray into pipe…

103ebde

…lineplan

fix split

9a8081c

fix and test

65b7aeb

Signed-off-by: Ubuntu <[email protected]>

[AIR/Docs] Add Predictor Docs (ray-project#25833)

445d38d

Signed-off-by: Ubuntu <[email protected]>

[air] Add _max_cpu_fraction_per_node to ScalingConfig and documentati…

41295f4

…on (ray-project#26634) Signed-off-by: Ubuntu <[email protected]>

[RLlib] improved unittests for dataset_reader and fixed bugs (ray-pro…

3d4af4a

…ject#26458) Signed-off-by: Ubuntu <[email protected]>

[RLlib]: Fix OPE trainables (ray-project#26279)

e999300

Co-authored-by: Kourosh Hakhamaneshi <[email protected]> Signed-off-by: Ubuntu <[email protected]>

[air] Add a warning if no CPUs are reserved for dataset execution (ra…

9faca47

…y-project#26643) Signed-off-by: Ubuntu <[email protected]>

added summary why and when to use bulk vs streaming data ingest (ray-…

d71114c

…project#26637) Signed-off-by: Ubuntu <[email protected]>

[AIR][CUJ] Make distributing training benchmark at silver tier (ray-p…

8594d20

…roject#26640) Signed-off-by: Ubuntu <[email protected]>

[Air][Data] Don't promote locality_hints for split (ray-project#26647)

78783dc

Why are these changes needed? Since locality_hints is an experimental feature, we stop promoting it in doc and don't enable it in AIR. See ray-project#26641 for more context Signed-off-by: Ubuntu <[email protected]>

Add Tao as Java worker code owner. (ray-project#26596)

3bb72c1

Signed-off-by: Ubuntu <[email protected]>

[RLlib] Add/reorder Args of Prioritized/MixIn MultiAgentReplayBuffer. (…

0d9216c

…ray-project#26428) Signed-off-by: Ubuntu <[email protected]>

[Serve] Remove EXPERIMENTAL inside the comments for user config (ray-…

5608c82

…project#26521) Signed-off-by: Ubuntu <[email protected]>

fix split

2d92130

Signed-off-by: Ubuntu <[email protected]>

fix and test

9e71304

Signed-off-by: Ubuntu <[email protected]>

jianoaix force-pushed the pipelineplan branch from 65b7aeb to 9e71304 Compare July 18, 2022 19:12

jianoaix requested a review from richardliaw as a code owner July 18, 2022 19:12

ericl requested changes Jul 19, 2022

View reviewed changes

ericl added the @author-action-required The PR author is responsible for the next step. Remove tag to send back to the reviewer. label Jul 19, 2022

jianoaix assigned c21 Jul 19, 2022

jianoaix added 3 commits July 19, 2022 22:27

feedback: naming/concept

84faf33

Merge branch 'master' of https://github.com/ray-project/ray into pipe…

397308f

…lineplan

feedback: run_by_pipeline->run_by_consumer

b4db0e6

ericl requested changes Jul 20, 2022

View reviewed changes

python/ray/data/_internal/block_list.py Outdated Show resolved Hide resolved

python/ray/data/_internal/plan.py Outdated Show resolved Hide resolved

feedback: make owned_by_consumer required keyword-only arg

9c41a71

jianoaix added 4 commits July 21, 2022 01:51

feedback: make run_by_consumer required keyword-only arg; and make th…

13c9ed6

…is arg in ExecutionPlan instead of Stage

typo

5d1ca73

undo no-op changes on stage_impl.py

0942498

undo continued

91b0b2f

ericl approved these changes Jul 21, 2022

View reviewed changes

fix

6d3e9aa

c21 reviewed Jul 21, 2022

View reviewed changes

jianoaix changed the title ~~Make execution stage/blocklist aware if it's run by Dataset or Datase…~~ Make execution plan/blocklist aware of the memory ownership and who runs the plan Jul 22, 2022

feedback: document transformation v.s. consumption concept

01f3da9

jianoaix removed the @author-action-required The PR author is responsible for the next step. Remove tag to send back to the reviewer. label Jul 22, 2022

ericl merged commit 8553df4 into ray-project:master Jul 22, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Make execution plan/blocklist aware of the memory ownership and who runs the plan #26650

Make execution plan/blocklist aware of the memory ownership and who runs the plan #26650

jianoaix commented Jul 17, 2022 •

edited

Loading

ericl left a comment

jianoaix commented Jul 21, 2022

c21 Jul 21, 2022

jianoaix Jul 22, 2022

c21 Jul 21, 2022

jianoaix Jul 22, 2022

c21 Jul 21, 2022

jianoaix Jul 21, 2022

c21 commented Jul 21, 2022 •

edited

Loading

jianoaix commented Jul 21, 2022

Make execution plan/blocklist aware of the memory ownership and who runs the plan #26650

Make execution plan/blocklist aware of the memory ownership and who runs the plan #26650

Conversation

jianoaix commented Jul 17, 2022 • edited Loading

Why are these changes needed?

Related issue number

Checks

ericl left a comment

Choose a reason for hiding this comment

jianoaix commented Jul 21, 2022

c21 Jul 21, 2022

Choose a reason for hiding this comment

jianoaix Jul 22, 2022

Choose a reason for hiding this comment

c21 Jul 21, 2022

Choose a reason for hiding this comment

jianoaix Jul 22, 2022

Choose a reason for hiding this comment

c21 Jul 21, 2022

Choose a reason for hiding this comment

jianoaix Jul 21, 2022

Choose a reason for hiding this comment

c21 commented Jul 21, 2022 • edited Loading

jianoaix commented Jul 21, 2022

jianoaix commented Jul 17, 2022 •

edited

Loading

c21 commented Jul 21, 2022 •

edited

Loading